#ifndef __MQ_CHANNEL_HPP__
#define __MQ_CHANNEL_HPP__

#include "mq_consumer.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
// #include "mq_router.hpp"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
// #include "mq_virtualhost.hpp"
#include "../mqcommon/mq_helper.hpp"
#include <mutex>
#include <condition_variable>
#include <unordered_map>

// cid信道ID,rid请求ID。

namespace mq
{
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;

    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;

        // Channel()
        // {
        //     DLOG("new Channel:%p",this);
        // }
        Channel(const muduo::net::TcpConnectionPtr& conn,
                const ProtobufCodecPtr& codec)
            :_cid(Helper::UuidHelper::uuid())
            ,_conn(conn)
            ,_codec(codec)
        {}
        ~Channel()
        {
            basicCancel();
        }
        std::string cid()
        {
            return _cid;
        }
        bool openChannel()
        {
            std::string rid = Helper::UuidHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);

            return resp->ok();
        }
        void closeChannel()
        {
            std::string rid = Helper::UuidHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        bool declareExchange(const std::string& ename,
                            ExchangeType type,
                            bool durable,
                            bool auto_delete,
                            google::protobuf::Map<std::string, std::string> &args
                )
        {
            std::string rid = Helper::UuidHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            
            return resp->ok();
        }
        void deleteExchange(const std::string& ename)
        {
            std::string rid = Helper::UuidHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        bool declareQueue(const std::string& queue_name,
                bool durable,
                bool exculsive,
                bool auto_delete,
                google::protobuf::Map<std::string,std::string>& args
            )
        {
            std::string rid = Helper::UuidHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);
            req.set_durable(durable);
            req.set_exculsive(exculsive);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);

            return resp->ok();
        }
        void deleteQueue(const std::string& queue_name)
        {
            std::string rid = Helper::UuidHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        bool queueBind(const std::string& ename,
                       const std::string& qname,
                       const std::string& key
            )
        {
            std::string rid = Helper::UuidHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_bind_key(key);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);

            return resp->ok();
        }
        void queueUnBind(const std::string& ename,
                        const std::string& qname
                            )
        {
            std::string rid = Helper::UuidHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        void basicPublish(const std::string& ename,
                          const BasicProperties* bp,
                          const std::string& body
                    )
        {
            std::string rid = Helper::UuidHelper::uuid();
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if(bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_rounting_key(bp->rounting_key());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            }

            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        void basicAck(const std::string& message_id)
        {
            if(_consumer.get() == nullptr)
            {
                DLOG("消息确认时，未找到消费者信息。");
                return ;
            }
            std::string rid = Helper::UuidHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(message_id);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        // 取消订阅
        void basicCancel()
        {
            if(_consumer.get() == nullptr)
            {
                // DLOG("取消订阅时，未找到消费者信息。");

                return ;
            }
            std::string rid = Helper::UuidHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->tag);
            req.set_queue_name(_consumer->qname);
            _codec->send(_conn,req);
            waitResponse(rid);

            return ;
        }
        bool basicConsume(const std::string& consumer_tag, 
                        const std::string& queue_name, 
                        bool auto_ack, 
                        const ConsumerCallback& cb)
        {
            // DLOG("构建consumer类。");
            if(_consumer.get() != nullptr)
            {
                DLOG("当前信道已订阅其他队列消息！");

                return false;
            }
            std::string rid = Helper::UuidHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
	        // consume还没有被创建出来。
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(queue_name);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);

            if(resp->ok() == false)
            {
                DLOG("添加订阅失败了!");

                return false;
            }
            
            // DLOG("制造consumer类。");
	        // _consumer对象直到这里才创建好。
            _consumer = std::make_shared<Consumer>(consumer_tag,queue_name,auto_ack,cb);
            DLOG("consumer tag:%s \t queue_name:%s \t ", consumer_tag.c_str(), queue_name.c_str());

            return true;
        }
    public:
        // 连接收到基础响应后，向hash_map中添加响应。
        void putBasicResponse(const basicCommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            // 插入新的需求。
            _basic_resp.insert(std::make_pair(resp->rid(),resp));
            // 唤醒线程。
            _cv.notify_all();

            return ;
        }
        // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理。
        void consume(const basicConsumeResponsePtr& resp)
        {
            if(_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者");

                return ;
            }
            if(_consumer->tag != resp->consumer_tag())
            {
                DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不一致！");

                return ;
            }
            _consumer->callback(resp->consumer_tag(),
                resp->mutable_properties(),
                resp->body());

            return ;
        }
    private:
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 满足条件就解锁。
            // 值引用、值捕捉。
            _cv.wait(lock,[&rid,this]()->bool
            {
                // 有元素了，就解锁。
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            // while(condition()) _cv.wait();
            // 虽然写是这么写，但是如果出现不存在rid对应的BasicCommonResponsePtr对象该怎么办。
            basicCommonResponsePtr basic_req = _basic_resp[rid];
            _basic_resp.erase(rid);

            return basic_req;
        }
    private:
        // 序列号。
        std::string _cid;
        // Tcp协议对象。
        muduo::net::TcpConnectionPtr _conn;
        // 
        ProtobufCodecPtr _codec;
        Consumer::Ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;
    };

    class ChannelManager
    {
    public:
        using Ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        Channel::Ptr create(const muduo::net::TcpConnectionPtr& conn,
                            const ProtobufCodecPtr& codec
                        )
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            Channel::Ptr channel = std::make_shared<Channel>(conn,codec);
            _channels.insert(std::make_pair(channel->cid(),channel));

            return channel;
        }
        void remove(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            _channels.erase(cid);

            return ;
        }
        Channel::Ptr get(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it != _channels.end())
            {
                return it->second;
            }

            return Channel::Ptr();
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::Ptr> _channels;
    };
}
#endif
